package com.atguigu.edu.app.dws;

import app.func.DimAsyncFunction;
import bean.TradeSourceOrderBean;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import util.DateFormatUtil;
import util.MyClickhouseUtil;
import util.MyKafkaUtil;

import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 *  交易域来源粒度下的各窗口统计
 */

public class DwsTradeSourceOrderWindow {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        String topic = "dwd_trade_order_detail";
        String group = "dws_trade_source_order_window";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getKafkaConsumer(topic, group));
        SingleOutputStreamOperator<TradeSourceOrderBean> mapDS = kafkaDS.map(jsonStr -> {
            JSONObject jsonObject = JSONObject.parseObject(jsonStr);
            String sourceId = jsonObject.getString("source_id");
            String orderId = jsonObject.getString("order_id");
            String userId = jsonObject.getString("user_id");
            Double finalAmount = jsonObject.getDouble("final_amount");
            Long ts = jsonObject.getLong("od_ts") * 1000L;
            return TradeSourceOrderBean.builder()
                    .sourceId(sourceId)
                    .orderIdSet(new HashSet<String>(
                            Collections.singleton(orderId)
                    ))
                    .userIdSet(new HashSet<String>(
                            Collections.singleton(userId)
                    ))
                    .orderAmount(finalAmount)
                    .ts(ts)
                    .build();
        });

        // TODO  设置水位线
        SingleOutputStreamOperator<TradeSourceOrderBean> withWatermarkDS = mapDS.assignTimestampsAndWatermarks(WatermarkStrategy.<TradeSourceOrderBean>forMonotonousTimestamps().withTimestampAssigner((value, ts) -> value.getTs()));

        // TODO 按照来源分组
        KeyedStream<TradeSourceOrderBean, String> keyedByProIdDS = withWatermarkDS.keyBy(TradeSourceOrderBean::getSourceId);

        // TODO 开窗
        WindowedStream<TradeSourceOrderBean, String, TimeWindow> windowDS = keyedByProIdDS.window(TumblingEventTimeWindows.of(Time.seconds(10L)));

        // TODO 聚合计算
        SingleOutputStreamOperator<TradeSourceOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<TradeSourceOrderBean>() {
                    @Override
                    public TradeSourceOrderBean reduce(TradeSourceOrderBean value1, TradeSourceOrderBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.getUserIdSet().addAll(value2.getUserIdSet());
                        value1.setOrderAmount(value1.getOrderAmount() + value2.getOrderAmount());

                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeSourceOrderBean, TradeSourceOrderBean, String, TimeWindow>() {
                    @Override
                    public void process(String s, Context context, Iterable<TradeSourceOrderBean> elements, Collector<TradeSourceOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(context.window().getStart());
                        String edt = DateFormatUtil.toYmdHms(context.window().getEnd());
                        for (TradeSourceOrderBean element : elements) {
                            element.setStt(stt);
                            element.setEdt(edt);
                            element.setOrderCount((long) element.getOrderIdSet().size());
                            element.setUserCount((long) element.getUserIdSet().size());
                            element.setTs(System.currentTimeMillis());
                            out.collect(element);
                        }
                    }
                }
        );
        // TODO 关联来源信息
        SingleOutputStreamOperator<TradeSourceOrderBean> fullInfoDS = AsyncDataStream.unorderedWait(reduceDS,
                new DimAsyncFunction<TradeSourceOrderBean>("dim_base_source".toUpperCase()) {
                    @Override
                    public void join(TradeSourceOrderBean obj, JSONObject dimJsonObj) throws Exception {
                        String sourceSite = dimJsonObj.getString("source_site".toUpperCase());
                        obj.setSourceSite(sourceSite);
                    }

                    @Override
                    public String getKey(TradeSourceOrderBean obj) {
                        return obj.getSourceId();
                    }
                },
                60 * 50, TimeUnit.SECONDS);


        // TODO 写入到OLAP数据库
        SinkFunction<TradeSourceOrderBean> sinkFunction = MyClickhouseUtil.<TradeSourceOrderBean>getSinkFunction(
                "\"insert into dws_trade_source_order_window values(?,?,?,?,?,?,?,?,?)\""

        );
        fullInfoDS.addSink(sinkFunction);
        env.execute();

    }
}
